package fr.unice.polytech.chord.core;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.UUID;

import org.apache.log4j.Logger;
import org.objectweb.proactive.Body;
import org.objectweb.proactive.InitActive;
import org.objectweb.proactive.RunActive;
import org.objectweb.proactive.Service;
import org.objectweb.proactive.api.PAActiveObject;
import org.objectweb.proactive.api.PAFuture;
import org.objectweb.proactive.core.body.exceptions.BodyTerminatedException;
import org.objectweb.proactive.core.body.request.Request;

import fr.unice.polytech.chord.core.messages.asynchronous.AddEntriesMessage;
import fr.unice.polytech.chord.core.messages.asynchronous.AddEntryMessage;
import fr.unice.polytech.chord.core.messages.asynchronous.AsynchronousMessage;
import fr.unice.polytech.chord.core.messages.asynchronous.GetEntryMessage;
import fr.unice.polytech.chord.core.messages.asynchronous.RemoveEntryMessage;
import fr.unice.polytech.chord.core.responses.asynchronous.ActionResponseMessage;
import fr.unice.polytech.chord.core.responses.asynchronous.AsynchronousResponseMessage;
import fr.unice.polytech.chord.core.responses.asynchronous.EntryResponseMessage;

/**
 * Represents a peer of type Chord.
 * 
 * The number of {@link ChordNode} that can be created is 2^{@link Key#NB_BITS}.
 * Notre that you must use {@link ChordFactory} in order to create a new
 * instance of a {@link ChordNode}.
 * 
 * @author Anthony Parisy
 * @author Laurent Pellegrino
 * @author Marc Valdener
 * 
 * @version 10/30/2009
 */
@SuppressWarnings("serial")
public class ChordNode implements Comparable<ChordNode>, InitActive, RunActive,
		Serializable {

	private static final Logger logger = Logger.getLogger(ChordNode.class);

	private static final int STABILIZE_INTERVAL_TIME = 300;

	private static final int TOLERANT_FAILURES = 3;

	private int fingerIndexToFix;

	private Key key;

	private ChordNode predecessor;

	private ChordNode successor;

	private FingerTable fingerTable;

	private final ChordNode[] successorsForTolerantFailures = new ChordNode[TOLERANT_FAILURES];

	private Entries<String> entries = new Entries<String>();

	private boolean alive = true;

	/**
	 * Constructor. An identifier will be automatically assigned.
	 */
	public ChordNode() {

	}

	/**
	 * Constructor.
	 * 
	 * @param id
	 *            the identifier to assign to the node.
	 */
	public ChordNode(Integer id) {
		if (id == null) {
			this.key = new Key(UUID.randomUUID().toString());
		} else {
			this.key = new Key(id);
		}
	}

	/**
	 * Creates new Chord network. It is called only by the first peer which want
	 * to create the network.
	 * 
	 * @return <code>true</code> if the creation has succeeded.
	 *         <code>false</code> otherwise.
	 */
	public boolean create() {
		logger.info("Node " + this.key + " has created a new network!");

		this.predecessor = null;
		this.successor = this.getStub();

		return true;
	}

	/**
	 * Joins a Chord ring with a node in the Chord ring.
	 * 
	 * @param node
	 *            a bootstrapping node.
	 */
	public boolean join(ChordNode node) {
		this.predecessor = null;
		this.successor = node.findSuccessor(this.key);

		logger.info("Node " + this.key + " has joined the network from node "
				+ node.getKey());

		return true;
	}

	/**
	 * Leaves the overlay by informing neighbors that it want to leave the
	 * network.
	 * 
	 * @return <code>true</code> if leave success, <code>false</code> otherwise.
	 */
	public boolean leave() {
		logger.info("Node " + this.key + " has left the network");

		this.alive = false;
		boolean res = false;

		if (this.equals(this.predecessor) && this.equals(this.successor)) {
			res = true;
		} else {
			PAFuture.waitFor(this.predecessor.setSuccessor(this.successor));
			PAFuture.waitFor(this.successor.setPredecessor(this.predecessor));

			// Gives all entries to successor
			res = ((ActionResponseMessage) PAFuture.getFutureValue(this.sendTo(
					this.successor, new AddEntriesMessage(this.entries
							.getEntries())))).hasSucceeded();
		}

		if (res) {
			PAActiveObject.terminateActiveObject(true);
		}

		return res;
	}

	private ChordNode findSuccessor(Key id, int index) {
		Key successorKey = null;

		try {
			successorKey = this.successor.getKey();
		} catch (Exception e) {
			if (index < this.successorsForTolerantFailures.length) {
				this.successor = this.successorsForTolerantFailures[index];
				return this.findSuccessor(id, index + 1);
			} else {
				this.successor = this.closestPrecedingNode(new Key(
						this.fingerTable.getFinger(index).getStartIndex()));
				return this.findSuccessor(id, index + 1);
			}
		}

		if (this.getStub() == this.successor) {
			return this.getStub();
		}

		if (id.isBetween(this.key, successorKey)
				|| id.compareTo(successorKey) == 0) {
			return this.successor;
		} else {
			ChordNode node = this.closestPrecedingNode(id);

			if (this.compareTo(node) == 0) {
				return this.getStub();
			}

			return node.findSuccessor(id);
		}
	}

	/**
	 * Lookups a successor of given key.
	 * 
	 * @param id
	 *            an identifier to lookup.
	 * @return the successor of given identifier.
	 */
	public ChordNode findSuccessor(Key id) {
		return this.findSuccessor(id, 0);
	}

	/**
	 * Lookups in the finger table the closest node which manages the key given
	 * in parameter.
	 * 
	 * @param id
	 *            the identifier used in order to lookup.
	 * @return the closest node which manages the specified key.
	 */
	public ChordNode closestPrecedingNode(Key id) {
		if (logger.isDebugEnabled()) {
			logger.debug(this.key.getValue() + ".closestPrecedingNode("
					+ id.getValue() + ")");
		}

		for (int i = Key.NB_BITS - 1; i >= 0; i--) {
			Finger finger = fingerTable.getFinger(i);

			try {
				Key fingerKey = finger.getNode().getKey();
				if (fingerKey.isBetween(this.key, id)) {
					return finger.getNode();
				}
			} catch (Exception e) {
				// wait next stabilization...
			}
		}

		return this.getStub();
	}

	/**
	 * Launch a new stabilization step in order to ensure that lookups execute
	 * correctly as the set of participating nodes changes. Chord must ensure
	 * that each node's successor pointer is up to date.
	 */
	public void update() {
		if (logger.isDebugEnabled()) {
			logger.debug("Node " + this.key
					+ " is performing stabilize algorithm");
		}
		this.stabilize();
		this.fixFingers();
		this.checkPredecessor();
	}

	/**
	 * Verifies the successor, and tells the successor about this node. It is
	 * called periodically (each {@code #STABILIZE_INTERVAL_TIME} ms).
	 * 
	 * @see #runActivity(Body)
	 */
	public void stabilize() {
		if (logger.isDebugEnabled()) {
			logger.debug("Node " + this.key + " is stabilizing...");
		}

		try {
			ChordNode node = (ChordNode) PAFuture.getFutureValue(this.successor
					.getPredecessor());

			if (node != null) {
				Key key = node.getKey();
				if (key.isBetween(this.key, this.successor.getKey())) {
					this.successor = node;
				}
			}

			PAFuture.waitFor(this.successor.notifyPredecessor(this.getStub()));
		} catch (Exception e) {
			// the successor is dead, wait for checkPredecessor...
		}
	}

	/**
	 * Called when the peer specified in parameter think it might be the
	 * predecessor of the current peer.
	 * 
	 * @param node
	 *            the new predecessor supposed.
	 * @return <code>true</code> if the call has succeeded, <code>false</code>
	 *         otherwise.
	 */
	public boolean notifyPredecessor(ChordNode node) {
		logger.debug("Node " + this.key + " is notifying predecessor...");
		try {
			Key nodeKey = node.getKey();

			this.updateSuccessorsForFaultTolerance();

			if (this.predecessor == null
					|| nodeKey.isBetween((Key) PAFuture
							.getFutureValue(this.predecessor.getKey()),
							this.key)) {
				this.predecessor = node;

				Map<Key, Entry<String>> entriesCheckingConstraints = new HashMap<Key, Entry<String>>();
				Iterator<Key> it = this.entries.getEntries().keySet()
						.iterator();
				Key currentKey;

				while (it.hasNext()) {
					currentKey = it.next();
					if (((this.key.compareTo(nodeKey) <= 0
							&& currentKey.compareTo(nodeKey) <= 0 && currentKey
							.compareTo(this.key) > 0) || (this.key
							.compareTo(nodeKey) > 0 && currentKey
							.compareTo(nodeKey) <= 0))) {
						entriesCheckingConstraints.put(currentKey, this.entries
								.getEntries().get(currentKey));
						it.remove();
					}
				}

				if (entriesCheckingConstraints.size() > 0) {
					PAFuture.waitFor(this.sendTo(node, new AddEntriesMessage(
							entriesCheckingConstraints)));
				}

			}
		} catch (Exception e) {
			// a peer is dead, wait for stabilization
		}

		return true;
	}

	/**
	 * Refreshes an entry of the finger table each time it is called. The size
	 * of the finger table is defined by the length of the {@link Key}.
	 */
	public void fixFingers() {
		logger.debug("Node " + this.key + " is fixing fingers...");

		this.fingerIndexToFix++;

		if (this.fingerIndexToFix >= Key.NB_BITS) {
			this.fingerIndexToFix = 0;
		}

		Finger finger = this.fingerTable.getFinger(this.fingerIndexToFix);
		ChordNode successorFound = (ChordNode) PAFuture.getFutureValue(this
				.findSuccessor(new Key(finger.getStartIndex())));
		finger.setNode(successorFound);
	}

	/**
	 * Checks if predecessor is always alive or not. If not, predecessor is set
	 * to <code>null</code> in order to be update in a next stabilization step..
	 */
	public void checkPredecessor() {
		logger.debug("Node " + this.key + " is checking predecessor...");

		if (this.predecessor != null
				&& !PAActiveObject.pingActiveObject(this.predecessor)) {
			this.predecessor = null;
		}
	}

	/**
	 * Update successors list for fault-tolerance.
	 */
	private void updateSuccessorsForFaultTolerance() {
		ChordNode successor = this;
		for (int index = 0; index < TOLERANT_FAILURES; index++) {
			successor = successor.getSuccessor();
			this.successorsForTolerantFailures[index] = successor;
		}
	}

	/**
	 * Puts a new {@link Entry} on the overlay.
	 * 
	 * @param key
	 *            the key associated to the data to add.
	 * @param value
	 *            the data to add.
	 * @return <code>true</code> if put has succeeded, <code>false</code>
	 *         otherwise.
	 */
	public boolean put(Entry<String> entry) {
		ChordNode node = this.findSuccessor(entry.getKey());

		if (logger.isInfoEnabled()) {
			logger.info("(" + entry.getKey() + "," + entry.getValue()
					+ ") has been added on node " + node.getKey());
		}

		return ((ActionResponseMessage) PAFuture.getFutureValue(this.sendTo(
				node, new AddEntryMessage(entry)))).hasSucceeded();
	}

	/**
	 * Lookups an {@link Entry} by using a given key.
	 * 
	 * @param key
	 *            the key to use for the lookup.
	 * @return the data found or <code>null</code> if the data is not on the
	 *         overlay.
	 */
	public Entry<String> get(Key key) {
		ChordNode node = this.findSuccessor(key);

		Entry<String> entryFound = ((EntryResponseMessage) PAFuture
				.getFutureValue(this.sendTo(node, new GetEntryMessage(key))))
				.getEntryFound();

		if (ChordNode.logger.isInfoEnabled()) {
			if (entryFound == null) {
				logger.info("Nobody on the network contains the key");
			} else {
				logger.info("Key " + key + " has been found on node "
						+ node.getKey() + " and associated value is '"
						+ entryFound + "'");
			}
		}

		return entryFound;
	}

	/**
	 * Removes a data by using the specified key.
	 * 
	 * @param key
	 *            the key to find.
	 * @return <code>true</code> if the remove has succeeded, <code>false</code>
	 *         otherwise.
	 */
	public boolean remove(Key key) {
		return ((ActionResponseMessage) PAFuture.getFutureValue(this.sendTo(
				this.findSuccessor(key), new RemoveEntryMessage(key))))
				.hasSucceeded();
	}

	/**
	 * Adds an entry.
	 * 
	 * @param entry
	 *            the entry to add.
	 * @return <code>true<code> if operation has succeeded. <code>false</code>
	 *         otherwise.
	 */
	public boolean addEntry(Entry<String> entry) {
		this.entries.getEntries().put(entry.getKey(), entry);
		return true;
	}

	/**
	 * Adds all the given entries.
	 * 
	 * @param entries
	 *            the entries to add.
	 * @return <code>true<code> if operation has succeeded. <code>false</code>
	 *         otherwise.
	 */
	public boolean addEntries(Map<Key, Entry<String>> entries) {
		this.entries.getEntries().putAll(entries);
		return true;
	}

	/**
	 * Return an entry by key.
	 * 
	 * @param key
	 *            the key to use in order to find the entry.
	 * @return the entry found or <code>null</code>.
	 */
	public Entry<String> getEntry(Key key) {
		return this.entries.getEntry(key);
	}

	/**
	 * Returns all the entries managed by the peer.
	 * 
	 * @return all the entries managed by the peer.
	 */
	public Entries<String> getEntries() {
		return this.entries;
	}

	/**
	 * Returns the number of entries that the peer is managing.
	 * 
	 * @return the number of entries that the peer is managing.
	 */
	public int getNbEntries() {
		return this.entries.getEntries().size();
	}

	/**
	 * Retrieves between to bounds.
	 * 
	 * @param from
	 *            lower bound.
	 * @param to
	 *            upper bound.
	 * @return the entry found for the given range.
	 */
	public Map<Key, Entry<String>> transfertEntries(Key from, Key to) {
		return this.entries.getEntriesBetween(from, to);
	}

	/**
	 * Removes an entry identified by a specified {@link Key}.
	 * 
	 * @param key
	 *            the key to use.
	 * @return <code>true<code> if operation has succeeded. <code>false</code>
	 *         otherwise.
	 */
	public boolean removeEntry(Key key) {
		return this.entries.getEntries().remove(key) != null;
	}

	/**
	 * Sends to a given peer the specified message.
	 * 
	 * @param node
	 *            the receiver of the message.
	 * @param msg
	 *            the message to send.
	 * @return a response associated to the type of the message sent.
	 */
	public AsynchronousResponseMessage sendTo(ChordNode node,
			AsynchronousMessage msg) {
		return node.receive(msg);
	}

	/**
	 * Receives a message.
	 * 
	 * @param msg
	 *            the message to receive.
	 * @return a response associated to the type of the message received.
	 */
	public AsynchronousResponseMessage receive(AsynchronousMessage msg) {
		return msg.handle(this);
	}

	/**
	 * Returns the key associated to this peer.
	 * 
	 * @return the key associated to this peer.
	 */
	public Key getKey() {
		return this.key;
	}

	/**
	 * Returns the predecessor of this peer.
	 * 
	 * @return the predecessor of this peer.
	 */
	public ChordNode getPredecessor() {
		return this.predecessor;
	}

	/**
	 * Returns the successor of this peer.
	 * 
	 * @return the successor of this peer.
	 */
	public ChordNode getSuccessor() {
		return this.successor;
	}

	/**
	 * Returns the finger table associated to this peer.
	 * 
	 * @return the finger table associated to this peer.
	 */
	public FingerTable getFingerTable() {
		return this.fingerTable;
	}

	/**
	 * Returns the remote instance of the current object.
	 * 
	 * @return the remote instance of the current object.
	 */
	public ChordNode getStub() {
		return (ChordNode) PAActiveObject.getStubOnThis();
	}

	/**
	 * Sets the predecessor.
	 * 
	 * @param predecessor
	 *            the new predecessor to use.
	 * @return <code>true<code> if operation has succeeded. <code>false</code>
	 *         otherwise.
	 */
	public boolean setPredecessor(ChordNode predecessor) {
		this.predecessor = predecessor;
		return true;
	}

	/**
	 * Sets the successor.
	 * 
	 * @param successor
	 *            the new successor to use.
	 * @return <code>true<code> if operation has succeeded. <code>false</code>
	 *         otherwise.
	 */
	public boolean setSuccessor(ChordNode successor) {
		this.successor = successor;
		return true;
	}

	/**
	 * Sets the finger table.
	 * 
	 * @param fingerTable
	 *            the new finger table to use.
	 */
	public void setFingerTable(FingerTable fingerTable) {
		this.fingerTable = fingerTable;
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public boolean equals(Object obj) {
		if (!(obj instanceof ChordNode)) {
			return false;
		}

		return this.compareTo((ChordNode) obj) == 0;
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public int hashCode() {
		return this.key.getValue().hashCode();
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public String toString() {
		StringBuffer buf = new StringBuffer();
		try {
			buf.append("Node ");
			buf.append(this.getKey().getValue());
			buf.append(" has P=");
			if (this.predecessor == null) {
				buf.append("null");
			} else {
				buf.append(this.predecessor.getKey());
			}
			buf.append(", S=");
			buf.append(this.successor.getKey());
			buf.append(", FT=[");
			for (int i = 0; i < Key.NB_BITS; i++) {
				Finger finger = fingerTable.getFinger(i);
				try {
					buf.append(finger.getNode().getKey());
				} catch (BodyTerminatedException e) {
					// just ignore it because peer has left and finger
					// table will be update automatically.
					buf.append("?");
				}
				if (i != Key.NB_BITS - 1) {
					buf.append(" ");
				}
			}
			buf.append("]");
			buf.append(", S=[");

			for (int j = 0; j < this.successorsForTolerantFailures.length; j++) {
				try {
					buf.append(this.successorsForTolerantFailures[j].getKey());
				} catch (Exception e) {
					buf.append("?");
				}

				if (j < this.successorsForTolerantFailures.length - 1) {
					buf.append(",");
				}
			}
			buf.append("]");
			buf.append(", E=[");

			int i = 0;
			for (Entry<String> entry : this.entries) {
				buf.append("(");
				buf.append(entry.getKey());
				buf.append(",");
				buf.append(entry.getValue());
				buf.append(")");
				if (i < this.entries.getEntries().size() - 1) {
					buf.append(", ");
				}
				i++;
			}
			buf.append("]");
		} catch (Exception e) {
			buf.append("?]");
		}

		return buf.toString();
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public int compareTo(ChordNode node) {
		return this.key.compareTo(node.getKey());
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public void initActivity(Body body) {
		PAActiveObject.setImmediateService("getKey");
		PAActiveObject.setImmediateService("getPredecessor");
		PAActiveObject.setImmediateService("getSuccessor");
		PAActiveObject.setImmediateService("findSuccessor");
		PAActiveObject.setImmediateService("notifyPredecessor");
		PAActiveObject.setImmediateService("receive");

		this.fingerTable = new FingerTable(this.getStub());

		logger.info("Node " + this.key + " is created");
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public void runActivity(Body body) {
		Service service = new Service(body);
		while (body.isActive()) {
			Request request = service
					.blockingRemoveOldest(ChordNode.STABILIZE_INTERVAL_TIME);
			// timeout has expired and no request to serve
			if (request == null && this.alive) {
				this.update();
			} else {
				service.serve(request);
			}
		}
	}

	/**
	 * Used in order to check is peer is alive.
	 * 
	 * @return <code>true</code> if peer is alive.
	 */
	public boolean isAware() {
		return true;
	}
}
